package com.spark.WorCount

import org.apache.spark.{SparkConf, SparkContext}

object TransformationOpsScala {

  def main(args: Array[String]): Unit = {
    val sparkContext=getSparkContext
    //1、map 算子操作--对集合中的每一个元素都扩大2倍
    //mapOp(sparkContext)

    //2、filter：过滤出集合中的偶数
    //filterOp(sparkContext)

    //3、flatMap：将行拆分为单词
    //flatMapOp(sparkContext)
    //4、groupByKey：对每个大区的主播进行分组
    //groupByKeyOp(sparkContext)
    //groupByKeyOp2(sparkContext)
    //6、reduceByKey：统计每个大区的主播数量
    //reduceByKeyOp(sparkContext)
    //7、sortByKey：对主播的音浪收入排序 默认是升序
    //sortByKeyOp(sparkContext)
    //8、join：打印每个主播的大区信息和音浪收入
    //joinOp(sparkContext)
    //9、distinct：统计当天开播的大区信息
    distinctOp(sparkContext)
    sparkContext.stop();
  }
  //对集合中的每一个元素都扩大2倍
  def mapOp(sc: SparkContext): Unit = {
    val rdd=sc.parallelize(Array(1,2,3,4,5,6))
    val newRdd=rdd.map(_ * 2)
    newRdd.foreach(println(_))
  }
  //过滤出集合中的偶数
  def filterOp(sc: SparkContext): Unit = {
    val rdd=sc.parallelize(Array(1,2,3,4,5,6,7))
    rdd.filter(_ % 2==0).foreach(println(_))
  }
  //将行数据拆分成单词
  def flatMapOp(sc: SparkContext) = {
    val rdd=sc.parallelize(Array("name age address","lz 20 haebin"))
    rdd.flatMap(_.split(" ")).foreach(println(_))
  }
  //根据大区信息area进行分组（两元组）
  def groupByKeyOp(sparkContext: SparkContext) = {
    //集合中的每一个元素是一个元组tuple
    val rdd=sparkContext.parallelize(Array(("15001","US"),("15002","CN"),("15003","CN"),("15004","USA"),("15005","US")))
    //map算子进行一对一转换（将tuple中的两个元素位置互换，保证第一个位置是键值进行分组操作）
    //groupByKey算子将集合中的元组数据根据第一列键值进行聚合
    rdd.map(tup=>(tup._2,tup._1)).groupByKey().foreach(tup=>{
      val area=tup._1
      print(area+":")
      val ids=tup._2
      for (id <- ids)
        print(id+" ")
      println()
    })
  }
  //根据大区信息area进行分组（三元组）
  def groupByKeyOp2(sparkContext: SparkContext) = {
    //集合中的每一个元素是一个元组tuple
    val rdd=sparkContext.parallelize(Array(("15001","US","male"),("15002","CN","female"),
      ("15003","CN","male"),("15004","USA","female"),("15005","US","male")))
    //map算子进行一对一转换（将tuple中的大区信息作为主键放在第一个位置，其他信息作为一个元组放在第二位置）
    //groupByKey算子将集合中的元组数据根据第一列键值进行分组
    rdd.map(tup=>(tup._2,(tup._1,tup._3))).groupByKey().foreach(tup=>{
      val area=tup._1
      print(area+" ")
      val it=tup._2;
      for ((id,male)<-it){
        print("<"+id+male+">")
      }
      println()
    })
  }
  //统计每个大区的主播数量
  def reduceByKeyOp(sparkContext: SparkContext) = {
    val rdd=sparkContext.parallelize(Array(("15001","US"),("15002","CN"),("15003","CN"),("15004","USA"),("15005","US")))
    //map算子 该需求只是用大区信息，因此map转换的时候只需要保留大区信息即可，计算数量的tuple格式（area,1）
    //reduceByKey算子 根据主键进行相加统计计算
    rdd.map(tup=>(tup._2,1)).reduceByKey(_ + _).foreach(println(_))
  }
  //根据音浪值进行排序（逆序输出--从小到大）
  def sortByKeyOp(sparkContext: SparkContext) = {
    val rdd=sparkContext.parallelize(Array(("15001",400),("1500",200),("15003",100),("15004",600),("15005",300)))
    rdd.map(tup=>(tup._2,tup._1)).sortByKey(false).foreach(println(_))
    //rdd.sortBy(_._2,false).foreach(println(_))
  }
  //打印每个主播的大区信息和音浪收入
  def joinOp(sparkContext: SparkContext) = {
    val dataRdd1=sparkContext.parallelize(Array(("15001","US"),("15002","CN"),("15003","CN"),("15004","USA"),("15005","US")))
    val dataRdd2=sparkContext.parallelize(Array(("15001",400),("1500",200),("15003",100),("15004",600),("15005",300)))
    //将两份数据根据主键进行合并，结果类型为tuple(key,(column1,column2,....))
    val dataRdd=dataRdd1.join(dataRdd2)
    dataRdd.foreach(tup=>{
      println(tup._1+"\t"+tup._2._1+"\t"+tup._2._2)
    })
  }
  //统计开播的大区信息
  def distinctOp(sparkContext: SparkContext) = {
    val dataRdd=sparkContext.parallelize(Array(("15001","US"),("15002","CN"),("15003","CN"),("15004","USA"),("15005","US")))
    dataRdd.map(_._2).distinct().foreach(println(_))
  }
  //获取SparkContext对象
  private def getSparkContext = {
    val conf = new SparkConf()
      .setAppName("TransformationOpsScala")
      .setMaster("local")
    new SparkContext(conf)
  }
}
